package com.ds.lens.flink.monitor.metricsjob;

import com.ds.lens.data.common.client.vo.LensMetricsTags;
import com.ds.lens.data.common.client.vo.LensMetricsTagsInVO;
import com.ds.lens.data.common.client.LensMonitorClient;
import com.ds.lens.data.common.client.LensResponseData;
import com.ds.lens.flink.monitor.MonitorJobContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.map.HashedMap;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.CollectionUtil;
import org.apache.flink.util.Collector;
import org.springframework.util.CollectionUtils;

import java.util.*;

@Slf4j
public class MetricsTagsProcessFunction extends ProcessWindowFunction<Metrics, Boolean, String, TimeWindow> {

    //key=appKey, value=Map<MD5(TagFull), metricName>
    private  Map<String, Map<String, String>> appKeyMetricsTagsCache;

    public MetricsTagsProcessFunction() {
        this.appKeyMetricsTagsCache = new HashedMap();
        LensMonitorClient monitorClient = MonitorJobContext.getInstance().getMonitorClient();
        LensResponseData<List<LensMetricsTags>> allMetricsTags = monitorClient.getAllMetricsTags();
        if(allMetricsTags != null){
            List<LensMetricsTags> metricsTagsData = allMetricsTags.getData();
            if (!CollectionUtils.isEmpty(metricsTagsData)) {
                for (LensMetricsTags m : metricsTagsData) {
                    String appKey = m.getAppKey();
                    if (appKeyMetricsTagsCache.containsKey(appKey)) {
                        appKeyMetricsTagsCache.get(appKey).put(Utils.convertMD5(m.getTagFull()), m.getMetricName());
                    } else {
                        Map<String, String> metricsTagsValueCache = new HashedMap();
                        metricsTagsValueCache.put(Utils.convertMD5(m.getTagFull()), m.getMetricName());
                        appKeyMetricsTagsCache.put(appKey, metricsTagsValueCache);
                    }
                }
            }
        }
    }

    @Override
    public void process(String key, Context context, Iterable<Metrics> elements, Collector<Boolean> out) {
        try {
            List<LensMetricsTagsInVO> metricsTagsInVOList = new ArrayList<>();
            Map<String, String> metricTags = appKeyMetricsTagsCache.get(key);
            if(elements.iterator().hasNext()){
                for (Metrics metrics : elements) {
                    Long appId = metrics.getAppId();
                    List<Metric> metricList = metrics.getMetrics();
                    if(!CollectionUtil.isNullOrEmpty(metricList)){
                        for(Metric metric : metricList){
                            String metricName = metric.getMetricName();
                            Map tags = metric.getTags();
                            if(metricTags !=null){
                                //DB有数据
                                handleDataWithCache(tags, metricTags, key, appId, metricName, metricsTagsInVOList);
                            }else {
                                //DB没有数据
                                metricTags = new HashMap();
                                appKeyMetricsTagsCache.put(key,metricTags);
                                handleDataWithoutCache(tags, metricTags, key, appId, metricName, metricsTagsInVOList);
                            }
                        }
                    }

                }
                //save
                MonitorJobContext.getInstance().getMonitorClient().saveMetricsTags(metricsTagsInVOList);
            }
        }catch (Exception e){
            log.error("MetricsTagsProcessFunction sink mysql process failed. ", e);
        }


    }

    private void handleDataWithCache(Map tags, Map<String, String> metricTags, String key, Long appId,  String metricName, List<LensMetricsTagsInVO> metricsTagsInVOList){

        if(!CollectionUtil.isNullOrEmpty(tags)){
            Set<Map.Entry<String, String>> entrySet = tags.entrySet();
            for(Map.Entry entry : entrySet){
                String tagFull = entry.getKey() + ":" + entry.getValue();
                if(!metricTags.containsValue(metricName)){
                    handleTagsData(metricTags, key, appId, metricName, metricsTagsInVOList, entry, tagFull);
                }else if(!metricTags.containsKey(Utils.convertMD5(tagFull))){
                    handleTagsData(metricTags, key, appId, metricName, metricsTagsInVOList, entry, tagFull);
                }
            }

        }else {
            //没有tags
            if(!metricTags.containsValue(metricName)){
                LensMetricsTagsInVO lensMetricsTagsInVO = new LensMetricsTagsInVO();
                lensMetricsTagsInVO.setAppKey(key);
                lensMetricsTagsInVO.setAppId(appId);
                lensMetricsTagsInVO.setMetricName(metricName);
                metricsTagsInVOList.add(lensMetricsTagsInVO);
                //更新缓存
                metricTags.put(Utils.convertMD5(lensMetricsTagsInVO.getTagFull()), metricName);
            }

        }
    }

    private void handleTagsData(Map<String, String> metricTags, String key, Long appId, String metricName, List<LensMetricsTagsInVO> metricsTagsInVOList, Map.Entry entry, String tagFull) {
        LensMetricsTagsInVO lensMetricsTagsInVO = new LensMetricsTagsInVO();
        lensMetricsTagsInVO.setAppKey(key);
        lensMetricsTagsInVO.setAppId(appId);
        lensMetricsTagsInVO.setMetricName(metricName);
        lensMetricsTagsInVO.setTagKey(entry.getKey().toString());
        lensMetricsTagsInVO.setTagValue(entry.getValue().toString());
        lensMetricsTagsInVO.setTagFull(tagFull);
        metricsTagsInVOList.add(lensMetricsTagsInVO);
        //更新缓存
        metricTags.put(Utils.convertMD5(tagFull), metricName);
    }

    private void handleDataWithoutCache(Map tags, Map<String, String> metricTags, String key, Long appId,  String metricName, List<LensMetricsTagsInVO> metricsTagsInVOList){

        if(!CollectionUtil.isNullOrEmpty(tags)){
            Set<Map.Entry<String, String>> entrySet = tags.entrySet();
            for(Map.Entry entry : entrySet){
                String tagFull = entry.getKey() + ":" + entry.getValue();
                handleTagsData(metricTags, key, appId,  metricName, metricsTagsInVOList, entry, tagFull);

            }

        }else {
            //没有tags
            LensMetricsTagsInVO lensMetricsTagsInVO = new LensMetricsTagsInVO();
            lensMetricsTagsInVO.setAppKey(key);
            lensMetricsTagsInVO.setAppId(appId);
            lensMetricsTagsInVO.setMetricName(metricName);
            metricsTagsInVOList.add(lensMetricsTagsInVO);
            //更新缓存
            metricTags.put(Utils.convertMD5(lensMetricsTagsInVO.getTagFull()), metricName);

        }
    }

}
